/**
 * Copyright (c) 2011-2015 1lifes.com
 * HSJ java-handlersocket
 * Id: HSClientHandler.java
 */
package com.elifes.hsj.client.netty;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.DefaultWriteCompletionEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.util.internal.ConcurrentHashMap;
import org.jboss.netty.util.internal.LinkedTransferQueue;

import com.elifes.hsj.client.HSJClient;
import com.elifes.hsj.packet.IPacket;
import com.elifes.hsj.util.IDGenerator;

/**
 * 描述：另一种处理方式 采用请求队列对应结果返回
 * @author yangqiang
 * @createtime 2011-11-12下午5:18:02
 *
 */
public class HSClientSeqHandler extends SimpleChannelUpstreamHandler{
	private static final Logger logger = Logger.getLogger(
			HSClientSeqHandler.class.getName());
	
	private BlockingQueue<IPacket> reqPacketQueue = new LinkedTransferQueue<IPacket>();
	private Executor reqPacketQueueExecutor = Executors.newCachedThreadPool();
	
	private HSJClient hsjClient;
	
	
	public HSClientSeqHandler(HSJClient hsjClient){
		this.hsjClient = hsjClient;
	}

	/* (non-Javadoc)
	 * @see org.jboss.netty.channel.SimpleChannelUpstreamHandler#messageReceived(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.MessageEvent)
	 */
	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent event)
			throws Exception {
		IPacket packet = reqPacketQueue.take();
		//TaskManager.getInstance().execute(event, hsjClient);
		TaskManager.getInstance().execute(event, packet);
	}

	/* (non-Javadoc)
	 * @see org.jboss.netty.channel.SimpleChannelUpstreamHandler#exceptionCaught(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.ExceptionEvent)
	 */
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
			throws Exception {
		//error handler requst queue
		IPacket packet = reqPacketQueue.take();
		super.exceptionCaught(ctx, e);
	}

	@Override
	public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
			throws Exception {
		if(e instanceof DefaultWriteCompletionEvent){//hs request
			final Channel channel = e.getChannel();
			reqPacketQueueExecutor.execute(new Runnable() {
				public void run() {
					IPacket packet = hsjClient.getPendingRequest().get(channel.getId());
					//删除中继map值
					hsjClient.removePendingRequest(channel.getId());
					reqPacketQueue.offer(packet);
					hsjClient.addChannelPool(packet.getIndexId(), channel);
				}
			});
			
			
		}
		super.handleUpstream(ctx, e);
	}

}
